package com.sjy.kafka;

/**
 * @author 苏健尧
 * @date 2024/8/23 17:19
 * @desciption: sd'
 */
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.sjy.websocket.MyWebSocketHandler;
import com.sjy.websocket.WsServerEndpoint;
import com.sjy.websocket.constant.WsConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.io.IOException;

@Slf4j
@Service
public class KafkaConsumer  {

    @Autowired
    private MyWebSocketHandler webSocketHandler;



    @Autowired
    private WsServerEndpoint wsServerEndpoint;

    @KafkaListener(topics = WsConstant.WS_TOPIC, groupId = "group_id")
    public void consume(String message) {
        log.info("接收到Kafka消息: " + message);
        // 处理接收到的消息
        String processedMessage = processMessage(message);

        // 将处理结果发送回相应的WebSocket客户端
        try {
            wsServerEndpoint.sendInfo(processedMessage, getSidFromMessage(message));
        } catch (IOException e) {
            log.error("发送处理结果失败", e);
        }
    }

    private String processMessage(String message) {
        // 在这里进行消息处理逻辑
        return "处理后的消息: " + message;
    }

    private String getSidFromMessage(String message) {
        // 从消息中提取sid
        JSONObject json_message = JSONUtil.parseObj(message);
        return json_message.getStr("sid");
    }
}
